package com.zhangyang.flink.cdc;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2023/2/6
 */
public class TableValueMain {


    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
//        configuration.setString(RestOptions.BIND_PORT, "8099");
        configuration.setString("rest.flamegraph.enabled", "true");
//        configuration.setString("classloader.resolve-order","parent-first");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);


        EnvironmentSettings envSettings = EnvironmentSettings
                .newInstance()

                .inStreamingMode()
//                .inBatchMode()
                .build();
//        env.setParallelism(3);
        // note: 增量同步需要开启CK
//        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, envSettings);
        tableEnvironment.executeSql(" CREATE TABLE asset_schema (\n" +
                " id STRING , " +
                "         `assets_id` STRING ,\n" +
                "          `ip_address` STRING, \n" +
                "          `host_name` STRING, \n" +
                "           primary key(id)  NOT ENFORCED" +
                "         ) WITH (\n" +
                "          'connector' = 'mysql-cdc',\n" +
                "          'hostname' = '10.215.10.103',\n" +
//                "          'hostname' = '127.0.0.1',\n" +
                "          'port' = '33060',\n" +
                "          'username' = 'totems',\n" +
                "          'password' = '123.Totems#@!',\n" +
                "          'database-name' = 'minke',\n" +
                "          'table-name' = 'nmap_assets_id'," +
                //  全量 + 增量同步
                "          'scan.startup.mode' = 'initial'      " +
                " )");
// 在 Table API 里不经注册直接“内联”调用函数
        Table select = tableEnvironment
                .from("asset_schema")
                .joinLateral(call(SplitLine4.class, $("ip_address")))
                .select($("single_ip_address"), $("assets_id"),$("host_name"));
        tableEnvironment.toChangelogStream(select).print("输出资产数据");

        env.execute("test1");
    }


    @FunctionHint(output = @DataTypeHint("ROW<single_ip_address STRING>"))
    public static class SplitLine4 extends TableFunction {



        public SplitLine4() {
        }

        public void eval(String str) {
            if(StringUtils.isNotEmpty(str)){
                for (String s : str.split(",")) {
                    // use collect(...) to emit a row
                    collect(Row.of(s));
                }
            }

        }
    }
}
